package data.platform.stream.internal.command;

import data.platform.common.service.command.RulePointCommandService;
import data.platform.common.service.command.RuleScheduleService;
import data.platform.common.service.query.MetricTagQueryService;
import data.platform.common.stream.aggregates.Rule;
import data.platform.common.stream.aggregates.TimeWindow;
import data.platform.common.stream.valueobjects.RuleState;
import data.platform.common.stream.valueobjects.WindowType;
import data.platform.common.util.DateUtil;
import data.platform.stream.internal.cache.RuleAggregateCache;
import data.platform.stream.internal.cache.RuleCache;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Mono;

import javax.annotation.PostConstruct;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.util.Date;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
@RequiredArgsConstructor
public class RuleScheduleServiceImpl implements RuleScheduleService {

    final MetricTagQueryService metricTagQueryService;

    final RuleCache ruleCache;

    final RuleAggregateCache ruleAggregateCache;

    final RulePointCommandService rulePointCommandService;

    ScheduledThreadPoolExecutor scheduledThreadPoolExecutor;

    Map<Rule, ScheduledFuture> scheduledFutureMap;

    private static final ThreadLocal<SimpleDateFormat> dateFormatter = ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS"));

    @PostConstruct
    public void init() {
        scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(10);
        // 任务可以从定时器中删除
        scheduledThreadPoolExecutor.setRemoveOnCancelPolicy(true);

        scheduledFutureMap = new ConcurrentHashMap<>();

        long activeRuleCount = ruleCache.getAllRules().stream()
                .filter(rule -> rule.getRuleState() == RuleState.ACTIVE)
                .peek(rule -> createScheduler(rule))
                .count();
        log.info("active rule count is: {}.", activeRuleCount);
    }

    public void createScheduler(Rule rule) {
        if (rule.getWindowType() == WindowType.MINUTE
                || rule.getWindowType() == WindowType.HOUR
                || rule.getWindowType() == WindowType.DAY) {
            startSchedule(rule);
        } else {
            startMonthAndYearSchedule(rule);
        }
    }

    @Override
    public Mono<Rule> createOrUpdateRule(Rule rule) {
        if (StringUtils.hasText(rule.getId())) {
            removeSchedule(rule);
        }
        rule = ruleCache.createOrUpdateRule(rule);
        createScheduler(rule);
        return Mono.just(rule);
    }

    @Override
    public Mono<String> startRule(String ruleId) {
        return Mono.just(ruleId)
                .map(id -> ruleCache.findRule(id))
                .filter(Optional::isPresent)
                .map(Optional::get)
                .doOnNext(this::removeSchedule)
                .doOnNext(this::createScheduler)
                .doOnNext(rule -> rule.setRuleState(RuleState.ACTIVE))
                .doOnNext(rule -> ruleCache.createOrUpdateRule(rule))
                .map(Rule::getId);
    }

    @Override
    public Mono<String> pauseRule(String ruleId) {
        return Mono.just(ruleId)
                .map(id -> ruleCache.findRule(id))
                .filter(Optional::isPresent)
                .map(Optional::get)
                .doOnNext(this::removeSchedule)
                .doOnNext(rule -> rule.setRuleState(RuleState.PAUSE))
                .doOnNext(rule -> ruleCache.createOrUpdateRule(rule))
                .map(Rule::getId);
    }

    public void removeSchedule(Rule rule) {
        if (scheduledFutureMap.containsKey(rule)) {
            ScheduledFuture scheduledFuture = scheduledFutureMap.get(rule);
            scheduledFuture.cancel(true);
            scheduledFutureMap.remove(rule);
        }
    }

    public void startSchedule(Rule rule) {
        RuleTask ruleTask = new RuleTask(rule, metricTagQueryService, ruleAggregateCache, rulePointCommandService, this);
        Long[] scheduleTime = scheduleTime(rule);
        ScheduledFuture scheduledFuture = scheduledThreadPoolExecutor.scheduleWithFixedDelay(ruleTask, scheduleTime[0], scheduleTime[1], TimeUnit.MILLISECONDS);
        //ScheduledFuture scheduledFuture = scheduledThreadPoolExecutor.schedule(ruleTask, scheduleTime[0], TimeUnit.MILLISECONDS);
        scheduledFutureMap.put(rule, scheduledFuture);
        log.info("ruleId:{}, startTime: {}, nextTime: {}", rule.getId(), dateFormatter.get().format(new Date().getTime() + scheduleTime[0]), dateFormatter.get().format(new Date().getTime() + scheduleTime[0] + scheduleTime[1]));
    }

    public void startMonthAndYearSchedule(Rule rule) {
        RuleTask ruleTask = new RuleTask(rule, metricTagQueryService, ruleAggregateCache, rulePointCommandService, this);
        Long[] scheduleTime = scheduleTime(rule);
        ScheduledFuture scheduledFuture = scheduledThreadPoolExecutor.schedule(ruleTask, scheduleTime[0], TimeUnit.MILLISECONDS);
        scheduledFutureMap.put(rule, scheduledFuture);
        log.info("ruleId:{}, startTime: {}.", rule.getId(), dateFormatter.get().format(new Date().getTime() + scheduleTime[0]));
    }

    // 任务第一次启动时间
    public Long[] scheduleTime(Rule rule) {
        Long[] scheduleTime = new Long[]{0L, 0L};

        LocalDateTime now = LocalDateTime.now();
        TimeWindow timeWindow = TimeWindow.getTimeWindow(rule.getWindowType(), rule.getWindowSize(), now);
        long delay = 0;

        if (rule.getWindowType() == WindowType.MINUTE) {
            delay = TimeUnit.MINUTES.toMillis(rule.getWindowSize());
        } else if (rule.getWindowType() == WindowType.HOUR) {
            delay = TimeUnit.HOURS.toMillis(rule.getWindowSize());
        } else if (rule.getWindowType() == WindowType.DAY) {
            delay = TimeUnit.DAYS.toMillis(rule.getWindowSize());
        }

        scheduleTime[0] = timeWindow.getEnd() - DateUtil.getDateOfLocalDateTime(now).getTime() + rule.getWaterMark() * 1000;
        scheduleTime[1] = delay;
        return scheduleTime;
    }
}
